// +------------------------------------------------
// | threadpool.c 
// | 线程 + 锁 + 队列 实现的线程池 
// | hatsusakana@gmail.com 
// +------------------------------------------------

#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#ifdef WIN32
#include <windows.h>
#include <process.h>
#else
#include <pthread.h>
#endif

#include "threadpool.h"

/** 统一一下pthread与window的锁接口 **/
#ifdef WIN32
#define THP_MUTEX_LOCK CRITICAL_SECTION
#define THP_INIT(x) InitializeCriticalSection(x)
#define THP_FREE(x) DeleteCriticalSection(x)
#define THP_LOCK(x) EnterCriticalSection(x)
#define THP_UNLOCK(x) LeaveCriticalSection(x)
#else
#define THP_MUTEX_LOCK pthread_mutex_t
#define THP_INIT(x) pthread_mutex_init(x, NULL)
#define THP_FREE(x) pthread_mutex_destroy(x)
#define THP_LOCK(x) pthread_mutex_lock(x)
#define THP_UNLOCK(x) pthread_mutex_unlock(x)
#endif

typedef struct ThreadPoolData {
    void (*cb)(void *);
    void *p;
    ThreadPool *pool;

    struct ThreadPoolData *next;
} ThreadPoolData;

struct ThreadPool {
    size_t runningNum;
    size_t maxThreaNum;
    size_t waitNum;
    size_t waitMaxNum;
    ThreadPoolData *head, *end;
    THP_MUTEX_LOCK lock;
};

static void thpool_over (ThreadPoolData *thPoolData);

/** 封装过的内部线程 **/
#ifdef WIN32
static DWORD WINAPI thpool_thread (LPVOID param) {
    ThreadPoolData *thPoolData = (ThreadPoolData *)param;
    thPoolData->cb(thPoolData->p);
    thpool_over(thPoolData);
    free(thPoolData);
    return TRUE;
}
#else
static void *thpool_thread (void *param) {
    ThreadPoolData *thPoolData = (ThreadPoolData *)param;
    thPoolData->cb(thPoolData->p);
    thpool_over(thPoolData);
    free(thPoolData);
}
#endif

static void _thpool_create_thread (void *p) {
#ifdef WIN32
    HANDLE handle = CreateThread(NULL, 0, thpool_thread, p, 0, 0);
    CloseHandle(handle);
#else
    pthread_t thid;
    pthread_attr_t attr;
    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
    pthread_create(&thid, &attr, thpool_thread, p);
#endif
}

/** 创建一个线程池 **/
ThreadPool *thpool_create (size_t num) {
    ThreadPool *pool = (ThreadPool *)calloc(1, sizeof(ThreadPool));
    if (!pool) return NULL;

    pool->runningNum = 0;
    pool->maxThreaNum = num;
    pool->waitNum = 0;
    pool->waitMaxNum = num * 2;
    pool->head = pool->end = NULL;
    THP_INIT(&pool->lock);
    return pool;
}

/** 设置缓存队列最大长度 **/
void thpool_setqueuenum (ThreadPool *pool, size_t num) {
    if (pool) pool->waitMaxNum = num;
}

/** 添加一条线程 **/
int thpool_join (ThreadPool *pool, void (*cb)(void *), void *p) {
    ThreadPoolData *thPoolData = NULL;

    if (!pool || !cb) return 0;

    THP_LOCK(&pool->lock);

    thPoolData = (ThreadPoolData *)calloc(1, sizeof(ThreadPoolData));
    if (!thPoolData) { THP_UNLOCK(&pool->lock); return 0; }
    thPoolData->cb = cb;
    thPoolData->p = p;
    thPoolData->pool = pool;
    thPoolData->next = NULL;

    // 线程池已满存入队列缓存
    if (pool->runningNum == pool->maxThreaNum) {
        
        // 队列缓存已满
        if (pool->waitNum >= pool->waitMaxNum) {
            THP_UNLOCK(&pool->lock);
            return 0;
        }

        if (pool->head == NULL) {
            pool->head = pool->end = thPoolData;
        } else {
            pool->end->next = thPoolData;
            pool->end = thPoolData;
        }

        pool->waitNum++;
    }

    // 直接运行线程
    else {
        pool->runningNum++;
        _thpool_create_thread(thPoolData);
    }

    THP_UNLOCK(&pool->lock);
    return 1;
}

/** 线程执行完毕清理内存,调用队列的下一条线程 **/
static void thpool_over (ThreadPoolData *thPoolData) {
    ThreadPool *pool = thPoolData->pool;

    THP_LOCK(&pool->lock);

    thPoolData->pool->runningNum--;
    if (thPoolData->pool->head != NULL) {
        ThreadPoolData *buf = thPoolData->pool->head;
        if (thPoolData->pool->head == thPoolData->pool->end)
            thPoolData->pool->head = thPoolData->pool->end = NULL;
        else thPoolData->pool->head = thPoolData->pool->head->next;

        pool->runningNum++;
        pool->waitNum--;
        _thpool_create_thread(buf);
    }

    THP_UNLOCK(&pool->lock);
}

static int thpool_isover(ThreadPool *pool) {
    int ret = 0;
    THP_LOCK(&pool->lock);
    ret = (pool->head != NULL || pool->runningNum != 0) ? 0 : 1;
    THP_UNLOCK(&pool->lock);
    return ret;
}

/** 等待所有线程运行完毕 **/
void thpool_wait (ThreadPool *pool) {
    if (!pool) return;
    while (thpool_isover(pool) == 0) {
#ifdef WIN32
        Sleep(1);
#else
        usleep(1000);
#endif
    }
}

/** 释放线程池(会等待所有线程释放) **/
void thpool_destroy (ThreadPool **pool) {
    if (!pool || !(*pool)) return;

    thpool_wait(*pool);
    THP_FREE(&((*pool)->lock));
    free(*pool);
    *pool = NULL;
}
